package org.alanlau;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * 控制器
 *
 * @author alan
 * @version 1.0
 * @date 2021/4/10 12:24
 */
@RestController
public class KafkaController {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 默认的发送消息
     */
    @RequestMapping("/send")
    public void send(@RequestParam String msg) {
        kafkaTemplate.send(TopicConstants.TOPIC_NAME, msg);
    }

    /**
     * 按照发送消息的key，进行hash，决定存到哪一个 partition
     *
     * @param key 消息key
     * @param msg 消息
     */
    @RequestMapping("/sendKey")
    public void sendKey(@RequestParam String key, @RequestParam String msg) {
        kafkaTemplate.send(TopicConstants.TOPIC_NAME, key, msg);
    }

    /**
     * 指定发送到一个 partition（不会管key的值是什么，一定会放到指定的分区中去）
     *
     * @param partition 分区
     * @param msg       消息
     */
    @GetMapping("/sendPartition")
    public void sendPartition(@RequestParam Integer partition, @RequestParam String msg) {
        kafkaTemplate.send(TopicConstants.TOPIC_NAME, partition, "key", msg);
    }

    /**
     * 发送到第二个TOPIC
     * @param msg 消息
     */
    @GetMapping("/send2")
    public void send2(@RequestParam String msg) {
        // 默认使用 round-robin 轮询添加到该 topic 下的各个 partition
        kafkaTemplate.send(TopicConstants.TOPIC_NAME2, msg);
    }
}
